package com.king.first.app.join;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author wdl
 * @Date 2022/6/7 14:07
 */
public class BroadcastConnectedStreamExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Integer> stream1 = env.fromElements(1,2,3);
        DataStream<Long> stream2 = env.fromElements(1L,2L,3L);

        MapStateDescriptor<String, Long> mapStateDescriptor = new MapStateDescriptor<>("broad-id", Types.STRING, Types.LONG);

        BroadcastStream<Long> broadcast = stream2.broadcast(mapStateDescriptor);

        SingleOutputStreamOperator<String> streamOperator = stream1.connect(broadcast)
                .process(new BroadcastProcessFunction<Integer, Long, String>() {
                    @Override
                    public void processElement(Integer value, BroadcastProcessFunction<Integer, Long, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        out.collect("stream1 : " + value);
                    }

                    @Override
                    public void processBroadcastElement(Long value, BroadcastProcessFunction<Integer, Long, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect("broadcast : " + value);
                    }
                });

        streamOperator.print();

        env.execute();
    }
}
